package consumer_test

import (
	"gitee.com/zaper/kafka-service/model"
	"github.com/petermattis/goid"
)

const (
	ExecFunc1 = "ExecFunc1"
	ExecFunc2 = "ExecFunc2"
)

func ReaderGroup(kafkaModel *model.KafkaConsumerModel) {
	switch kafkaModel.StartFunc {
	case ExecFunc1:
		execReaderMessGroup(kafkaModel)
	case ExecFunc2:
		execConsumerMessGroup(kafkaModel)
	default:
		return
	}
}

func execReaderMessGroup(kafkaModel *model.KafkaConsumerModel) {
	for i := 0; i < kafkaModel.CustomerCount; i++ {
		go func() {
			//defer sw.Done()
			// 初始化每个协程的信号量
			kafkaModel.StopChan[goid.Get()] = make(chan struct{})
			execReaderMess(kafkaModel)
		}()
	}
}

func execConsumerMessGroup(kafkaModel *model.KafkaConsumerModel) {
	for i := 0; i < kafkaModel.CustomerCount; i++ {
		go func() {
			//defer sw.Done()
			// 初始化每个协程的信号量
			kafkaModel.StopChan[goid.Get()] = make(chan struct{})
			execConsumerMess(kafkaModel)
		}()
	}
}

func deleteSignMap(m map[int64]chan struct{}, key int64) {
	sw.Lock()
	delete(m, key)
	sw.Unlock()
}
